package com.mytijian.task.zookeeper;

import com.google.common.base.Charsets;
import com.mytijian.task.lifecycle.ApplicationLifycycle;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.transaction.CuratorTransactionFinal;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

public class ZookeeperRegistryCenter extends ApplicationLifycycle{
    private Logger logger = LoggerFactory.getLogger(ZookeeperRegistryCenter.class);
    private CuratorFramework client;
    private ZookeeperConfig zookeeperConfig;
    private Map<String, TreeCache> map = new ConcurrentHashMap<>();
    public CuratorFramework getClient() {
        return client;
    }
    public ZookeeperRegistryCenter(ZookeeperConfig zookeeperConfig) {
        this.zookeeperConfig = zookeeperConfig;
    }
    @Override
    public void init () {
        CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                .connectString(zookeeperConfig.getConnectionString())
                .retryPolicy(new ExponentialBackoffRetry(1000, 2, 3000))
                .namespace(zookeeperConfig.getNameSpace());
//        if (0 != zkConfig.getSessionTimeoutMilliseconds()) {
//            builder.sessionTimeoutMs(zkConfig.getSessionTimeoutMilliseconds());
//        }
//        if (0 != zkConfig.getConnectionTimeoutMilliseconds()) {
//            builder.connectionTimeoutMs(zkConfig.getConnectionTimeoutMilliseconds());
//        }
//        if (!Strings.isNullOrEmpty(zkConfig.getDigest())) {
//            builder.authorization("digest", zkConfig.getDigest().getBytes(Charsets.UTF_8))
//                    .aclProvider(new ACLProvider() {
//
//                        @Override
//                        public List<ACL> getDefaultAcl() {
//                            return ZooDefs.Ids.CREATOR_ALL_ACL;
//                        }
//
//                        @Override
//                        public List<ACL> getAclForPath(final String path) {
//                            return ZooDefs.Ids.CREATOR_ALL_ACL;
//                        }
//                    });
//        }
        client = builder.build();
        client.start();
        try {
            if (!client.blockUntilConnected(3000 * 2, TimeUnit.MILLISECONDS)) {
                client.close();
                throw new RuntimeException("连接超时");
            }
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
            ex.printStackTrace();
        }
    }

    @Override
    public int initSort() {
        return 1;
    }

    @Override
    public void destory() {
        logger.info("zookeeper 连接关闭");
        for (Map.Entry<String, TreeCache> each : map.entrySet()) {
            each.getValue().close();
        }
        waitForCacheClose();
        CloseableUtils.closeQuietly(client);
    }

    @Override
    public int destorySort() {
        return 0;
    }

    public void executeInTransaction(final TransactionExecutionCallback callback) {
        try {
            CuratorTransactionFinal curatorTransactionFinal = getClient().inTransaction().check().forPath("/").and();
            callback.execute(curatorTransactionFinal);
            curatorTransactionFinal.commit();
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
        }
    }
    /* TODO 等待500ms, cache先关闭再关闭client, 否则会抛异常
     * 因为异步处理, 可能会导致client先关闭而cache还未关闭结束.
     * 等待Curator新版本解决这个bug.
     * BUG地址：https://issues.apache.org/jira/browse/CURATOR-157
     */
    private void waitForCacheClose() {
        try {
            Thread.sleep(500L);
        } catch (final InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }
    public void persist(final String key, final String value) {
        try {
            if (!isExisted(key)) {
                client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(key, value.getBytes(Charsets.UTF_8));
            } else {
                update(key, value);
            }
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
        }
    }
    public void update(final String key, final String value) {
        try {
            client.inTransaction().check().forPath(key).and().setData().forPath(key, value.getBytes(Charsets.UTF_8)).and().commit();
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
        }
    }
    public TreeCache addCache(String cachePath) {
        TreeCache cache = new TreeCache(client, cachePath);
        try {
            cache.start();
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
            ex.printStackTrace();
            throw new RuntimeException("添加Tree缓存失败");
        }
        map.put(cachePath + "/", cache);
        return cache;
    }
    public void addListener(String cachePath, TreeCacheListener treeCacheListener) {
        TreeCache treeCache = map.get(cachePath +"/");
        if (treeCache == null) {
            treeCache = addCache(cachePath);
        }
        treeCache.getListenable().addListener(treeCacheListener);
    }
    public void persistEphemeral(final String key, final String value) {
        try {
            if (isExisted(key)) {
                client.delete().deletingChildrenIfNeeded().forPath(key);
            }
            client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key, value.getBytes(Charsets.UTF_8));

            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
            ex.printStackTrace();
        }
    }
    public boolean isExisted(final String key) {
        try {
            return null != client.checkExists().forPath(key);
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
            ex.printStackTrace();
            return false;
        }
    }
    public void remove(final String key) {
        try {
            client.delete().deletingChildrenIfNeeded().forPath(key);
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
        }
    }
    public String get(final String key) {
        TreeCache cache = findTreeCache(key);
        if (null == cache) {
            return getDirectly(key);
        }
        ChildData resultInCache = cache.getCurrentData(key);
        if (null != resultInCache) {
            return null == resultInCache.getData() ? null : new String(resultInCache.getData(), Charsets.UTF_8);
        }
        return getDirectly(key);
    }
    public String getDirectly(final String key) {
        try {
            return new String(client.getData().forPath(key), Charsets.UTF_8);
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
            ;
            return null;
        }
    }
    private TreeCache findTreeCache(final String key) {
        for (Map.Entry<String, TreeCache> entry : map.entrySet()) {
            if (key.startsWith(entry.getKey())) {
                return entry.getValue();
            }
        }
        return null;
    }
    public List<String> getChildrenKeys(final String key) {
        try {
            List<String> result = client.getChildren().forPath(key);
            Collections.sort(result, new Comparator<String>() {

                @Override
                public int compare(final String o1, final String o2) {
                    return o2.compareTo(o1);
                }
            });
            return result;
            //CHECKSTYLE:OFF
        } catch (final Exception ex) {
            //CHECKSTYLE:ON
            return Collections.emptyList();
        }
    }
}
